package com.kitten.count.consumer;

import com.google.common.collect.Maps;
import com.kitten.count.constant.MQConstants;
import com.kitten.count.constant.RedisKeyConstants;
import com.kitten.count.enums.FollowUnfollowTypeEnum;
import com.kitten.count.model.dto.CountFollowUnfollowMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;


/**
 * @author kitten
 */

@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kitten_group_" + MQConstants.TOPIC_COUNT_FOLLOWING,
        topic = MQConstants.TOPIC_COUNT_FOLLOWING
)
public class CountFollowingConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, String> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;



    @Override
    public void onMessage(String body) {
        log.info("Received message: " + body);
        if (StringUtils.isEmpty(body)) {
            return;
        }
        // 关注数和粉丝数计数场景不同，单个用户无法短时间内关注大量用户，所以无需聚合
        CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = JsonUtils.parseObject(body, CountFollowUnfollowMqDTO.class);
        // 操作类型
        Integer type = countFollowUnfollowMqDTO.getType();
        // 用户ID
        Long userId = countFollowUnfollowMqDTO.getUserId();

        String redisKey = RedisKeyConstants.buildCountUserKey(userId);
        Boolean isExist = redisTemplate.hasKey(redisKey);
        if (isExist) {  // 存在
            int count = Objects.equals(type, FollowUnfollowTypeEnum.FOLLOW.getCode()) ? 1 : -1; // 关注 或 取关
            redisTemplate.opsForHash().increment(redisKey, RedisKeyConstants.FIELD_FOLLOWING_TOTAL, count);    // count:user:2, key:following_total, value:1 | -1
        }
        // 写库
        Message<String> message = MessageBuilder.withPayload(body).build(); // 消息传递的数据
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FOLLOWING_2_DB, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 关注countService统计结果落库成功, {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 关注countService统计结果落库失败, {}", throwable);
            }
        });
    }
}
